package com.hub.realtime.flinkshims.core;


import com.hub.realtime.flinkshims.core.enums.SqlErrorType;
import com.hub.realtime.flinkshims.core.model.SqlError;
import lombok.extern.slf4j.Slf4j;
import org.apache.calcite.config.Lex;
import org.apache.calcite.sql.parser.SqlParser;
import org.apache.calcite.sql.validate.SqlConformance;
import org.apache.flink.sql.parser.validate.FlinkSqlConformance;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.planner.calcite.CalciteConfig;
import org.apache.flink.table.planner.delegation.FlinkSqlParserFactories;
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
import org.apache.flink.table.planner.utils.TableConfigUtils;
import org.apache.flink.util.ExceptionUtils;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.List;

import static com.hub.realtime.flinkshims.core.SqlCommand.CREATE_VIEW;
import static com.hub.realtime.flinkshims.core.enums.SqlErrorType.VERIFY_FAILED;


@Slf4j
public class FlinkSqlValidator {

    /**
     * 构建flink sql解析器
     *
     * @return
     */
    private static Object parser(ClassLoader classLoader,String plannerClass) throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
        TableConfig tableConfig = StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment(), EnvironmentSettings.newInstance().build()).getConfig();
        SqlParser.Config sqlParserConfig = JavaScalaConversionUtil.<SqlParser.Config>toJava(getCalciteConfig(tableConfig).getSqlParserConfig()).orElseGet(
                () -> {
                    SqlConformance conformance = getSqlConformance(tableConfig);
                    return SqlParser
                            .config()
                            .withParserFactory(FlinkSqlParserFactories.create(conformance))
                            .withConformance(conformance)
                            .withLex(Lex.JAVA)
                            .withIdentifierMaxLength(256);
                }
        );

        Class czl = classLoader.loadClass(plannerClass);
        Constructor constructor = czl.getConstructor(SqlParser.Config.class);
        return constructor.newInstance(sqlParserConfig);

    }


    private static CalciteConfig getCalciteConfig(TableConfig tableConfig) {
        return TableConfigUtils.getCalciteConfig(tableConfig);
    }

    private static FlinkSqlConformance getSqlConformance(TableConfig tableConfig) {
        SqlDialect sqlDialect = tableConfig.getSqlDialect();
        switch (sqlDialect) {
            case HIVE:
                return FlinkSqlConformance.HIVE;
            case DEFAULT:
                return FlinkSqlConformance.DEFAULT;
            default:
                throw new TableException("Unsupported SQL dialect: " + sqlDialect);
        }
    }


    /**
     * 校验sql合法性
     *
     * @param sql
     * @return
     */
    public static <T> SqlError verifySql(String sql, String plannerClass , ClassLoader classLoader) {

        try {
            SqlError sqlError = null;
            List<SqlCommandCall> callsqlCommands = SqlCommandParser.parseSQL(sql);
            for (SqlCommandCall call : callsqlCommands) {
                String args = call.getOperands()[0];
                SqlCommand command = call.getCommand();
                String last = call.getOperands()[call.getOperands().length - 1];

                switch (command) {
                    case SET:
                        if (!FlinkSqlExecutor.tableConfigOptions().containsKey(args)) {
                            sqlError = new SqlError(SqlErrorType.SYNTAX_ERROR, args.concat(" is not a valid table/sql config")
                                    , sql.replaceFirst(";|$", ";"));
                            return sqlError;
                        }
                        break;
                    case RESET:
                        if (args != "ALL" && !FlinkSqlExecutor.tableConfigOptions().containsKey(args)) {
                            sqlError = new SqlError(SqlErrorType.SYNTAX_ERROR,
                                    args.concat(" is not a valid table/sql config"),
                                    sql.replaceFirst(";|$", ";"));
                            return sqlError;
                        }
                        break;
                    case SHOW_CATALOGS:
                    case SHOW_CURRENT_CATALOG:
                    case SHOW_DATABASES:
                    case SHOW_CURRENT_DATABASE:
                    case SHOW_TABLES:
                    case SHOW_VIEWS:
                    case SHOW_FUNCTIONS:
                    case SHOW_MODULES:
                    case CREATE_FUNCTION:
                    case CREATE_CATALOG:
                    case CREATE_TABLE:
                    case CREATE_VIEW:
                    case CREATE_DATABASE:
                    case DROP_CATALOG:
                    case DROP_DATABASE:
                    case DROP_TABLE:
                    case DROP_VIEW:
                    case DROP_FUNCTION:
                    case ALTER_DATABASE:
                    case ALTER_TABLE:
                    case ALTER_FUNCTION:
                    case USE:
                    case USE_CATALOG:
                    case SELECT:
                    case INSERT_INTO:
                    case INSERT_OVERWRITE:
                    case EXPLAIN:
                    case DESC:
                    case DESCRIBE:
                        try {
                          Object obj=  parser(classLoader,plannerClass);
                            Class clz= obj.getClass();

                            Method[] declaredMethods = ((Class<?>) clz).getDeclaredMethods();

                            Method method = null;
                           for(Method m:declaredMethods)
                           {
                               if(m.getName().equals("parse")){
                                   method=m;
                               }
                           }
                            if (command == CREATE_VIEW) {
                                method.invoke(obj,last);
                            } else {
                                method.invoke(obj,args);
                            }
                        } catch (Exception ex) {
                            log.error("verify error:".concat(ExceptionUtils.stringifyException(ex)));
                            sqlError = new SqlError(SqlErrorType.SYNTAX_ERROR,
                                    ex.getCause().getLocalizedMessage(),
                                    args.trim().replaceFirst(";|$", ";"));
                            return sqlError;
                        }
                        break;
                    default:
                        sqlError = new SqlError(SqlErrorType.UNSUPPORTED_SQL, "unsupported sql",
                                sql.replaceFirst(";|$", ";"));
                        return sqlError;

                }
            }
            return null;
        } catch (Exception ex) {
            log.error("verify error:".concat(ExceptionUtils.stringifyException(ex)));
            SqlError sqlError = new SqlError();
            sqlError.setSqlErrorType(VERIFY_FAILED);
            sqlError.setException(ex.getMessage());
            return sqlError;
        }


    }


}
